﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using EasyNetQ;
using EasyNetQ.AutoSubscribe;
using EasyNetQ.Consumer;
using Newtonsoft.Json;

namespace Nl.RabbitMq.Demo.EasyNetQWrapper
{
    /// <summary>
    /// RabbitMQ客户端封装类，基于EasyNetQ。
    /// <para>
    /// <example>
    /// 使用方法：
    /// <code>
    /// using(var mq = new RabbitMqClient('rabbitmq连接字符串'))
    /// { ...
    /// }
    /// </code>
    /// </example>
    /// </para>
    /// </summary>
    public class RabbitMqClient : IDisposable
    {
        private readonly IBus _bus;

        #region 发布/订阅

        /// <summary>
        /// 消息订阅
        /// </summary>
        /// <param name="subscriptionId">消息订阅标识</param>
        /// <param name="process">
        /// 消息处理委托方法
        /// <para>
        /// <example>
        /// 例如：
        /// <code>
        /// message=>Task.Factory.StartNew(()=>{
        ///     Console.WriteLine(message);
        /// })
        /// </code>
        /// </example>
        /// </para>
        /// </param>
        public void Subscribe<TMessage>(string subscriptionId, Action<TMessage> process) where TMessage : class
        {
            _bus.Subscribe<TMessage>(subscriptionId, message =>
            {
                try
                {
                    process(message);
                }
                catch (Exception ex)
                {
                    _bus.Send<MqExcepotionMessage>("MQErrorQueue", new MqExcepotionMessage(JsonConvert.SerializeObject(message), ex));
                }
            });
        }

        /// <summary>
        /// 消息订阅
        /// </summary>
        /// <param name="subscriptionId">消息订阅标识</param>
        /// <param name="process">
        /// 消息处理委托方法
        /// <para>
        /// <example>
        /// 例如：
        /// <code>
        /// message=>Task.Factory.StartNew(()=>{
        ///     Console.WriteLine(message);
        /// })
        /// </code>
        /// </example>
        /// </para>
        /// </param>
        /// <param name="topic">topic</param>
        public void SubscribeWithTopic<TMessage>(string subscriptionId, Action<TMessage> process, string topic)
            where TMessage : class
        {
            _bus.Subscribe<TMessage>(subscriptionId, message =>
                {
                    try
                    {
                        process(message);
                    }
                    catch (Exception ex)
                    {
                        _bus.Send<MqExcepotionMessage>("MQErrorQueue", new MqExcepotionMessage(JsonConvert.SerializeObject(message), ex));
                    }
                },
                c =>
                {
                    if (topic != null)
                    {
                        c.WithTopic(topic);
                    }
                });
        }

        /// <summary>
        /// 自动订阅
        /// </summary>
        /// <param name="assemblyName"></param>
        /// <param name="subscriptionIdPrefix"></param>
        /// <param name="topic"></param>
        public void AutoSubscribe(string assemblyName, string subscriptionIdPrefix, string topic)
        {
            var subscriber = new AutoSubscriber(_bus, subscriptionIdPrefix);
            if (!string.IsNullOrEmpty(topic))
                subscriber.ConfigureSubscriptionConfiguration = x => x.WithTopic(topic);
            subscriber.Subscribe(Assembly.Load(assemblyName));
        }

        /// <summary>
        /// 发布一条消息(广播)
        /// </summary>
        /// <param name="message"></param>
        public void Publish<TMessage>(TMessage message) where TMessage : class
        {
            _bus.PublishAsync(message).ContinueWith(OnPushMessagePushed);
        }

        /// <summary>
        /// 指定Topic，发布一条消息
        /// </summary>
        /// <param name="message"></param>
        /// <param name="topic"></param>
        public void PublishWithTopic<TMessage>(TMessage message, string topic) where TMessage : class
        {
            if (string.IsNullOrEmpty(topic))
                Publish(message);
            else
                _bus.PublishAsync(message, x => x.WithTopic(topic)).ContinueWith(OnPushMessagePushed);
        }

        #endregion


        #region 消息入列确认逻辑

        /// <summary>
        /// 当推送到队列失败
        /// </summary>
        protected virtual void OnPushMessagePushed(Task pushTask)
        {
            //发送确认
            if (pushTask.IsCompleted)
            {
                //todo  消息入列成功逻辑（日志？）
            }

            if (pushTask.IsFaulted)
            {
                //todo  消息入列失败逻辑（日志？错误处理？）
                Console.Out.WriteLine(pushTask.Exception);
            }
        }

        #endregion

        #region

        #endregion

        /// <summary>
        /// 同步方式给指定队列发送一条信息（无法确认是否入列）
        /// </summary>
        /// <param name="queue">队列名称</param>
        /// <param name="message">消息</param>
        public void Send<TMessage>(string queue, TMessage message) where TMessage : class
        {
            _bus.Send(queue, message);
        }

        /// <summary>
        /// 给指定队列批量发送信息
        /// </summary>
        /// <param name="queue">队列名称</param>
        /// <param name="messages">消息</param>
        public void SendMany<TMessage>(string queue, IList<TMessage> messages) where TMessage : class
        {
            foreach (var message in messages)
            {
                SendAsync(queue, message);
            }
        }


        /// <summary>
        /// 给指定队列发送一条信息（异步）
        /// </summary>
        /// <param name="queue">队列名称</param>
        /// <param name="message">消息</param>
        /// <returns></returns>
        public async void SendAsync<TMessage>(string queue, TMessage message) where TMessage : class
        {
            await _bus.SendAsync(queue, message).ContinueWith(OnPushMessagePushed);
        }

        /// <summary>
        /// 从指定队列接收一条信息，并做相关处理。
        /// </summary>
        /// <param name="queue">队列名称</param>
        /// <param name="process">
        /// 消息处理委托方法
        /// <para>
        /// <example>
        /// 例如：
        /// <code>
        /// message=>Task.Factory.StartNew(()=>{
        ///     Console.WriteLine(message);
        /// })
        /// </code>
        /// </example>
        /// </para>
        /// </param>
        public void Receive<TMessage>(string queue, Action<TMessage> process) where TMessage : class
        {
            _bus.Receive<TMessage>(queue, onMessage: message =>
            {
                try
                {
                    process(message);
                }
                catch (Exception ex)
                {
                    _bus.Send<MqExcepotionMessage>("MQErrorQueue", new MqExcepotionMessage(JsonConvert.SerializeObject(message), ex));
                }

                ;
            });
        }

        #region 构造/析构

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="connectionString">rabbitmq连接字符串</param>
        public RabbitMqClient(string connectionString)
        {
            if (string.IsNullOrEmpty(connectionString))
                throw new ArgumentNullException(nameof(connectionString));

            ConnectionConfiguration connectionConfiguration = new ConnectionConfiguration()
            {
                UseBackgroundThreads = true,
                PrefetchCount = 1000,
                Hosts = new List<HostConfiguration>() {new HostConfiguration() {Host = "localhost", Port = 5672}}
            };
            _bus = RabbitHutch.CreateBus(connectionConfiguration, x => x.Register<IEasyNetQLogger, CustomMqLogger>());
            //
            //_bus.Advanced.Container.Resolve<IConventions>().ErrorExchangeNamingConvention = info => "MyExchangeNaming";
            //_bus.Advanced.Container.Resolve<IConventions>().ErrorQueueNamingConvention = () => "MyErrorQueueNaming";
        }

        /// <summary>
        /// 资源释放
        /// </summary>
        public void Dispose()
        {
            _bus?.Dispose();
        }

        #endregion
    }
}